package com.yungam.influxdb.service;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.QueryApi;
import com.influxdb.client.WriteApi;
import com.influxdb.query.FluxTable;
import com.yungam.influxdb.config.DbConfig;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;

@Slf4j
@Service
public class InfluxdbCommonService {

    @Resource
    private DbConfig config;
    @Getter
    private InfluxDBClient influxDBClient;
    @Getter
    private WriteApi writeApi;
    @Getter
    private QueryApi queryApi;

    @PostConstruct
    public void init() {
        influxDBClient = InfluxDBClientFactory.create(
                config.getInflux().getUrl(),
                config.getInflux().getToken().toCharArray(),
                config.getInflux().getOrg(),
                config.getInflux().getBucket());
        writeApi = influxDBClient.getWriteApi();
        queryApi = influxDBClient.getQueryApi();
    }

    /*
     * 根据 ip 查询对应的数据
     * */
    public List<FluxTable> queryByIp(String ip, Long[] range, Integer interval) {
        if (interval == null) interval = 30;
        String queryStr = "from(bucket:\"" + config.getInflux().getBucket() + "\")" +
                "|> range(start: " + range[0] + ", stop: " + range[1] + ")" +
                "|> filter(fn: (r) => r[\"_measurement\"] == \"flink_data\")" +
                "|> filter(fn:(r)=> r.ip == \"" + ip + "\")" +
                "|> aggregateWindow(every: " + interval +  "s, fn: mean, createEmpty: false) " +
                "|> yield(name: \"mean\")";
        try {
            return queryApi.query(queryStr);
        } catch (Exception e) {
            log.error("读取错误, 重新创建QueryApi", e);
            queryApi = influxDBClient.getQueryApi();
            return queryApi.query(queryStr);
        }
    }

    /*
     * 计算对应采集项的电流标准差
     * */
    public List<FluxTable> queryCurrentStdValue(String ip, Integer collectorId, Long[] range) {
        String queryStr = "from(bucket:\"" + config.getInflux().getBucket() + "\")" +
                "|> range(start: " + range[0] + ", stop: " + range[1] + ")" +
                "|> filter(fn: (r) => r[\"_measurement\"] == \"flink_data\")\n" +
                "|> filter(fn: (r) => r[\"ip\"] == \"" + ip + "\" and r[\"id\"] == \"" + collectorId + "\")\n" +
                "|> filter(fn: (r) => r[\"_value\"] >= 5)\n" +
                "|> stddev()";
        log.info("influxdb ===== queryStr: {}", queryStr);
        try {
            return queryApi.query(queryStr);
        } catch (Exception e) {
            log.error("读取错误, 重新创建QueryApi", e);
            queryApi = influxDBClient.getQueryApi();
            return queryApi.query(queryStr);
        }
    }

    /*
     * 计算对应采集项的电压标准差
     * collectorIds: [电流, 电压]
     * 不管算电流还是电压标准差, 都需要根据电流是否大于起弧来筛选
     * */
    public List<FluxTable> queryVoltageStdValue(String ip, Integer[] collectorIds, Long[] range) {
        String queryStr = "from(bucket:\"" + config.getInflux().getBucket() + "\")\n" +
                "|> range(start: " + range[0] + ", stop: " + range[1] + ")\n" +
                "|> filter(fn: (r) => r[\"_measurement\"] == \"flink_data\")\n" +
                "|> filter(fn: (r) => r[\"ip\"] == \"" + ip + "\")\n" +
                "|> filter(fn: (r) => r[\"id\"] == \"" + collectorIds[0] + "\" or r[\"id\"] == \"" + collectorIds[1] + "\")\n" +
                "|> keep(columns: [\"_time\", \"_value\", \"id\"])\n" +
                "|> pivot(rowKey: [\"_time\"], columnKey: [\"id\"], valueColumn: \"_value\")\n" +
                "|> filter(fn: (r) => r[\"" + collectorIds[0] + "\"] >= 5)\n" +
                "|> stddev(column: \"" + collectorIds[1] + "\")";
        log.info("influxdb ===== queryStr: {}", queryStr);
        try {
            return queryApi.query(queryStr);
        } catch (Exception e) {
            log.error("电压标准差查询失败", e);
            queryStr = "from(bucket:\"" + config.getInflux().getBucket() + "\")\n" +
                    "|> range(start: " + range[0] + ", stop: " + range[1] + ")\n" +
                    "|> filter(fn: (r) => r[\"_measurement\"] == \"flink_data\")\n" +
                    "|> filter(fn: (r) => r[\"ip\"] == \"" + ip + "\" and r[\"id\"] == \"" + collectorIds[1] + "\")\n" +
                    "|> filter(fn: (r) => r[\"_value\"] >= 5)\n" +
                    "|> stddev(column: \"" + collectorIds[1] + "\")";
            log.info("重新查询 queryStr: {}", queryStr);
            return queryApi.query(queryStr);
        }
    }

}
